package com.atguigu.day01;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

public class Example1 {
    // 不要忘记抛出异常！
    public static void main(String[] args) throws Exception {
        // 获取流执行环境，类似spark中获取Spark Context
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 定义了处理数据的有向无环图
        env
                // 从socket读取数据并输出，输出的类型是String
                .socketTextStream("localhost", 9999)
                // 设置数据源算子`socketTextStream`的并行子任务的数量为1
                .setParallelism(1)
                // map阶段: "hello world" -> ("hello", 1); ("world", 1)
                // 由于是1对多的转换，所以这里使用flatMap算子
                // 外部类
                .flatMap(new Tokenizer())
                // flatMap的并行子任务的数量为1
                .setParallelism(1)
                // shuffle阶段，按照key也就是单词将数据进行分组
                // f0表示元组的第1个字段，相当于scala中的`_1`
                // keyBy的作用是为每一条输入数据指定key
                // 将输入数据的key设置为f0字段
                // KeySelector<输入数据的泛型，key的泛型>
                // 匿名类
                .keyBy(new KeySelector<Tuple2<String, Integer>, String>() {
                    // 参数：输入数据
                    @Override
                    public String getKey(Tuple2<String, Integer> input) throws Exception {
                        return input.f0;
                    }
                })
                // reduce阶段
                // 每个key都会维护一个累加器
                // 外部类
                .reduce(new WordCount())
                // `reduce`任务的并行子任务的数量也是1
                .setParallelism(1)
                // 将聚合结果输出
                .print()
                // `print`的并行子任务的数量是1
                .setParallelism(1);

        // 提交并执行程序
        env.execute();
    }

    // ReduceFunction<输入数据的泛型>
    public static class WordCount implements ReduceFunction<Tuple2<String, Integer>> {
        // reduce方法定义的是输入数据和累加器的聚合规则
        // 累加器是输入数据的key所对应的累加器，因为每个key都会维护自己的累加器
        // 累加器最开始为空
        // 返回值是聚合以后的新的累加器
        // reduce算子会将累加器输出到下游
        // reduce算子的输入，输出和累加器的类型是一样的
        @Override
        public Tuple2<String, Integer> reduce(Tuple2<String, Integer> accumulator, Tuple2<String, Integer> input) throws Exception {
            return Tuple2.of(accumulator.f0, accumulator.f1 + input.f1);
        }
    }

    // FlatMapFunction<输入的泛型，输出的泛型>
    public static class Tokenizer implements FlatMapFunction<String, Tuple2<String, Integer>> {
        // 第一个参数：输入数据
        // 第二个参数：集合，用来收集将要发送的数据
        @Override
        public void flatMap(String input, Collector<Tuple2<String, Integer>> out) throws Exception {
            // 使用空格切分字符串
            String[] words = input.split(" ");
            // 遍历单词数组，将每个单词都转换成元组，并收集到集合中
            for (String word : words) {
                // collect方法将待发送数据收集到集合当中
                // flink会自动将集合中的数据发送下去
                // `Tuple2.of`实例化一个2元组
                out.collect(Tuple2.of(word, 1));
            }
        }
    }
}
